当前单年交易表只能支持描述性汇总，以下业务问题需要额外字段与独立依据：
本案例使用朝阳医院2018年药品销售数据，完成四大分析任务：
| 任务编号 | 分析内容 | 核心方法 |
|---|---|---|
| 任务1 | 销售最好/最差的药品 | 分组聚合 + 柱状图 |
| 任务2 | 销售额时间变化趋势 | 时间序列 + 折线图 |
| 任务3 | 各月份最佳药品 | 循环 + 分组统计 |
| 任务4 | 应收—实收差额的描述性分类 | 特征工程 + 频率统计 |
分析的第一步是导入必要的库并读取数据：
pandas：数据分析核心库numpy：数值计算库seaborn / matplotlib：可视化库pd.read_excel() 读取 Excel 数据文件df.shape 和 df.head() 预览数据结构原始数据需要进行以下预处理操作：
dropna() 删除含缺失值的行duplicated().sum() 统计重复记录| 阶段 | 学习步骤 | 应得到什么 |
|---|---|---|
| 1 数据要求 | 核对授权、期间、商品编码与金额单位 | 数据字典与样本量 |
| 2 质量检查 | 检查日期、缺失、重复与金额关系 | 清洗核对表 |
| 3 描述任务 | 计算排名、趋势和差额分类 | 带分母的结果表 |
| 4 决策边界 | 区分描述统计与政策因果 | 负责人、判断条件和再次检查周期 |
df、df_sum、df_sum_n10、values1 的业务含义、数据类型或取值范围，并判断哪一个输入最可能改变结果。# 注：朝阳医院2018年销售数据.xlsx数据文件本地没有，但平台已经内置
# ⚠️ 平台原始代码 - 请原样输入至教学平台（注释除外），平台才会判定答案正确
import pandas as pd # 导入Pandas数据分析库
import numpy as np # 导入NumPy数值计算库
import seaborn as sns # 导入Seaborn可视化库
import matplotlib.pyplot as plt # 导入Matplotlib绘图库
plt.rcParams["font.sans-serif"] = ["SimHei"] # 设置Matplotlib全局参数
#数据读取
df = pd.read_excel("朝阳医院2018年销售数据.xlsx")
#数据预览与预处理
## 查看数据维度
print(df.shape)
print(df.head()) # 输出前几行数据
## 查看数据信息
print(df.info())
df['社保卡号']=df['社保卡号'].astype(str) # 转换数据类型
df['商品编码']=df['商品编码'].astype(str) # 转换数据类型
df['购药日期']=df['购药时间'].str.split(pat=' ',expand=True)[0] # 对字符串列进行处理
df['购药星期']=df['购药时间'].str.split(pat=' ',expand=True)[1] # 对字符串列进行处理
## 查看各列的缺失值
df.isnull().sum()
## 缺失值处理
## 直接删除
df.dropna(axis=0,inplace=True)
## 查看重复值数量
df.duplicated().sum()
#探索性数据分析
##任务1：销售情况最好的药品是？最差的药品是？
##提示：通过对药品进行分组聚合统计，在此处我们评定销售情况好的标准是销售数量总和。之后取出前n个药品进行可视化展示
df_sum=df.groupby('商品名称')['实收金额'].sum().reset_index() # 按指定列分组聚合
print(df_sum.nlargest(2,'实收金额')) #用于获取实收金额列中最大的两个值所对应的行。
print(df_sum.nsmallest(2,'实收金额')) #用于获取实收金额列中最小的两个值所对应的行。
df_sum_n10=df_sum.nlargest(10,'实收金额') # 筛选出排名前10的记录
plt.figure(figsize=(15,10)) # 创建图形画布
plt.grid() # 显示网格线
plt.bar(df_sum_n10['商品名称'],df_sum_n10['实收金额']) # 绘制柱状图
plt.savefig("1.png") # 保存图形至文件
plt.close() # 关闭文件或连接
#任务2：整体销售额的时间变化趋势
##数据集中存在时间数据，但是不够简洁，所以需要先对时间数据进行处理：
df["购药时间_月"] = df["购药时间"].map(lambda x:x.split(" ")[0].split("-")[1]) ##购药日期所属月份
df["购药时间_星期"] = df["购药时间"].map(lambda x:x.split(" ")[1]) ##购药当天是星期几？
#对时间数据处理完成之后，就可以进行时间维度上的销售情况分析，与任务1一致，评定指标也选择为销售数量之和。
# 在下方开始你的分析
df['购药日期_月']=df['购药日期'].str.split(pat='-',expand=True)[1]
plt.figure(figsize=(15,10)) # 创建图形画布
plt.grid() # 显示网格线
sns.lineplot(x='购药日期_月',y='实收金额',data=df,estimator='sum') # 绘制折线图
plt.savefig("2.png") # 保存图形至文件
plt.close() # 关闭文件或连接
plt.figure(figsize=(15,10)) # 创建图形画布
plt.grid() # 显示网格线
sns.lineplot(x='购药星期',y='实收金额',data=df,estimator='sum') # 绘制折线图
plt.savefig("3.png") # 保存图形至文件
plt.close() # 关闭文件或连接
#任务3：各月份卖的最好的药品是那些？
values1=df['购药日期_月'].unique()
values1=pd.Series(values1) # 创建Series序列values1
for value in values1: # 遍历values1中的每个value
df_value=df[df['购药日期_月']==value] # 提取购药日期_月列作为df_value变量
df_print=df_value.groupby(['购药日期_月','商品名称'])['实收金额'].sum().nlargest(1).reset_index() # 按指定列分组聚合
print(df_print) # 输出数据框数据
#任务4：描述应收金额与实收金额存在差额的交易占比
df['差额']=df['应收金额']-df['实收金额']
df['是否社保减免药品']=df['差额'].apply(lambda x : '是' if x!=0 else '否') # 定义匿名函数df['是否社保减免药品']
df['是否社保减免药品'].value_counts(normalize=True) # 统计是否社保减免药品列各取值的频率分布
#####总结
# 运行后按药品报告销量、样本量与期间；当前不预填最好药品
# 运行后按药品报告销量、样本量与期间；当前不预填最差药品
# 时间关系需在实际月度输出、供货与促销口径上核对
# 应收—实收差额只做描述分类，不据此推断医保使用或购买意愿df_print 行缩进少于循环体、随后 print 又恢复缩进；按标准 Python 解析会报 IndentationError。import pandas as pd # 导入表格工具解析日期并聚合
def monthly_top_product(pharmacy_sales): # 定义去标识化月度冠军接口
required = {'购药时间', '商品编码', '商品名称', '实收金额'} # 声明最小必要字段
if not required <= set(pharmacy_sales.columns): # 检查字段要求
raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 缺列时停止
safe_sales = pharmacy_sales.loc[:, sorted(required)].copy() # 仅保留最小必要字段
safe_sales['购药月份'] = pd.to_datetime(safe_sales['购药时间'], errors='coerce').dt.to_period('M') # 解析月份
if safe_sales['购药月份'].isna().any(): # 检查日期解析结果
raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 日期异常时停止
monthly_product = safe_sales.groupby(['购药月份', '商品编码', '商品名称'], as_index=False)['实收金额'].sum() # 汇总月度商品实收
monthly_top = monthly_product.sort_values(['购药月份', '实收金额'], ascending=[True, False]).groupby('购药月份').head(1) # 每月取最高实收商品
return monthly_top # 返回便于核对结果运行后核对：核对 df、df_sum、df_sum_n10、values1 是否按预测参与运算，实际输出是否与预测一致；若不一致，先检查类型、单位、索引/字段和运算顺序。
拓展练习：只改变一个关键输入或业务场景，先预测输出如何变化，再运行验证并解释变化原因。
目标：找出销售额最高和最低的药品
核心方法：
groupby('商品名称') 按药品分组实收金额 列调用 .sum() 求和nlargest(n) 提取销售前 n 名nsmallest(n) 提取销售后 n 名
# 提取销售金额TOP10的药品
df_sum_n10 = df_sum.nlargest(10, '实收金额')
plt.figure(figsize=(15, 10))
plt.bar(df_sum_n10['商品名称'], df_sum_n10['实收金额'], color='steelblue')
plt.title('销售金额TOP10药品', fontsize=16)
plt.xlabel('药品名称', fontsize=12)
plt.ylabel('销售金额(元)', fontsize=12)
plt.xticks(rotation=45, ha='right') # 旋转标签防止重叠
plt.grid(axis='y', alpha=0.3)
plt.tight_layout()
plt.show()运行平台数据后，应按以下规则解读：
目标：揭示销售额的月度趋势和星期趋势
核心方法：
str.split('-')[1]sns.lineplot() 绘制折线图estimator='sum' 对每组销售额求和
# 从购药日期提取月份
df['购药日期_月'] = df['购药日期'].str.split(pat='-', expand=True)[1]
# 绘制月度销售趋势
plt.figure(figsize=(15, 10))
sns.lineplot(x='购药日期_月', y='实收金额', data=df,
estimator='sum', errorbar=None)
plt.title('各月销售额变化趋势', fontsize=16)
plt.xlabel('月份', fontsize=12)
plt.ylabel('销售金额(元)', fontsize=12)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()目标：找出每个月销售额最高的药品
核心方法：
unique() 获取所有月份的唯一值for 循环遍历每个月份groupby + nlargest(1) 找出每月冠军先按“月份×药品”报告样本量、销量和销售额，再检查缺货、价格、促销与节假日。只有连续多个周期出现可核对模式，才把季节性列为备货假设；未经药学与供应链共同核对，不据此提前加库存。
目标：描述应收与实收存在差额的交易占比，不识别医保因果效应
核心思路：
当平台数据、字段字典和合规授权齐备时，按下表写出待验证结果：
| 分析维度 | 关键发现 |
|---|---|
| 销售排名 | 药品、销售额/量、患者人次、期间与缺货边界 |
| 时间趋势 | 月/星期分组的样本量与实际图表，不作无依据归因 |
| 季节假设 | 用多年同期、流行病与供货记录另行验证 |
| 差额分类 | 只做描述性占比；医保政策效应需独立识别策略 |
本案例涉及的 Python 数据分析核心技能：
groupby() + sum() / count()nlargest() / nsmallest()str.split() 提取日期特征apply(lambda) 构造新变量plt.bar() 柱状图 / sns.lineplot() 折线图case 文件；中国本地解答绑定 stock_basic_data 与 financial_statement，字段 ts_code/name/industry/area/end_date/revenue/operate_profit/accounts_receivimport pandas as pd # 导入表格库
import matplotlib.pyplot as plt # 导入绘图库生成练习要求的TOP10图
def summarize_platform_pharmacy(pharmacy_sales): # 定义平台药店任务完整接口
required={'购药时间','商品编码','商品名称','应收金额','实收金额'} # 对齐受保护任务实际中文字段
if not required<=set(pharmacy_sales.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 缺列时暂停分析
clean_sales=pharmacy_sales.dropna(subset=list(required)).drop_duplicates().copy() # 清除关键缺失与完全重复行
if clean_sales.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 空样本时停止
clean_sales['购药日期']=pd.to_datetime(clean_sales['购药时间'].astype(str).str.split().str[0],errors='coerce') # 解析平台日期部分
if clean_sales['购药日期'].isna().any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 日期无法解析时停止
clean_sales['差额']=clean_sales['应收金额']-clean_sales['实收金额'] # 计算描述性收付差额
top_products=clean_sales.groupby(['商品编码','商品名称']).agg(rows=('实收金额','size'),received=('实收金额','sum')).nlargest(10,'received') # 按平台实际评定指标生成TOP10
top10_chart,top10_axis=plt.subplots() # 创建TOP10药品柱状图画布
top10_plot=top_products.reset_index().sort_values('received') # 整理为水平柱状图的递增顺序
top10_axis.barh(top10_plot['商品名称'],top10_plot['received']) # 按实收金额绘制TOP10水平柱
top10_axis.set(title='实收金额TOP10药品',xlabel='实收金额',ylabel='药品') # 标明图表口径与坐标轴
top10_chart.tight_layout() # 压紧图表边距避免药品名称裁切
monthly=clean_sales.set_index('购药日期').resample('ME').agg(rows=('实收金额','size'),received=('实收金额','sum'),amount_gap=('差额','sum')) # 生成月度趋势
gap_groups=clean_sales.assign(gap_status=clean_sales['差额'].gt(0).map({True:'有差额',False:'无正差额'})).groupby('gap_status').agg(rows=('商品编码','size'),amount_gap=('差额','sum')) # 汇总差额分类及分母
if len(monthly)<2 or gap_groups['rows'].sum()==0: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 期间或分母不足时停止
return top_products,top10_chart,monthly,gap_groups # 返回TOP10表与图、趋势和差额输出def select_latest_disclosure(frame,value_fields): # 统一财报披露版本选择约定
required={'ts_code','end_date',*value_fields}; versions=[c for c in ['ann_date','f_ann_date'] if c in frame.columns] # 固定键与可接受版本字段
if not required<=set(frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 缺列时停止
work=frame.dropna(subset=list(required)).copy(); work['end_date']=pd.to_datetime(work['end_date'],errors='coerce') # 清理必需字段
for column in versions: work[column]=pd.to_datetime(work[column],errors='coerce') # 解析披露日期
keys=['ts_code','end_date']; work=work.dropna(subset=['end_date']); duplicate_mask=work.duplicated(keys,keep=False); before=int(duplicate_mask.sum()) # 统计选择前重复
if before and (not versions or work.loc[duplicate_mask,versions].isna().all(axis=1).any()): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 无版本依据时停止
work['_version']=work[versions].max(axis=1) if versions else pd.NaT # 合并双版本字段为排序键
latest_version=work.groupby(keys)['_version'].transform('max') if versions else pd.Series(pd.NaT,index=work.index) # 为每个公司报告期定位最晚披露时点
latest_rows=work.loc[~duplicate_mask | work['_version'].eq(latest_version)].copy() # 仅保留单例或并列最晚版本
tied_mask=latest_rows.duplicated(keys,keep=False); tied_latest_rows=int(tied_mask.sum()) # 统计并列最晚版本行
conflicting_latest_keys=int(latest_rows.loc[tied_mask].groupby(keys,dropna=False)[value_fields].nunique(dropna=False).gt(1).any(axis=1).sum()) if tied_latest_rows else 0 # 统计并列且数值冲突的披露键
if conflicting_latest_keys: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 并列最新值冲突时请先检查数据后再继续
selected=latest_rows.drop_duplicates(keys+value_fields).drop_duplicates(keys).drop(columns='_version',errors='ignore') # 仅合并数值完全一致的并列行
after=int(selected.duplicated(keys).sum()); audit={'duplicate_rows_before':before,'tied_latest_rows':tied_latest_rows,'conflicting_latest_keys':conflicting_latest_keys,'duplicate_rows_after':after,'version_fields':versions} # 返回并列与冲突检查
if after: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 选择后仍重复则停止
return selected,audit # 返回唯一版本与检查依据from pathlib import Path # 导入路径工具绑定规定资产
import pandas as pd # 导入表格库以标准化HDF字段
import numpy as np # 导入有限值检查工具
root=Path('/home/ubuntu/r2_data_mount/data/stock') # 指定股票数据目录
basic_path,financial_path=root/'stock_basic_data.h5',root/'financial_statement.h5' # 绑定两张可用HDF快照
if not basic_path.exists() or not financial_path.exists(): raise FileNotFoundError('未找到课程股票数据文件，请从课程数据下载入口获取') # 提示读者下载文件
basic=pd.read_hdf(basic_path,key='stock_basic_info',columns=['order_book_id','symbol','industry_name','province']).rename(columns={'order_book_id':'raw_code','symbol':'name','industry_name':'industry','province':'area'}) # 只读取公司筛选字段并标准化
basic['ts_code']=basic['raw_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False) # 统一证券代码后缀
basic['area']=basic['area'].astype('string').str.replace(r'[省市]$','',regex=True) # 统一省市简称
basic_fields={'ts_code','name','industry','area'} # 定义基础表字段结构
financial_fields={'ts_code','end_date','revenue','operate_profit','accounts_receiv'} # 定义财报字段结构
pharma=basic.loc[basic['industry'].str.contains('医药|生物|制药',na=False) & basic['area'].isin(['上海','江苏','浙江','安徽'])] # 筛选长三角医药公司
if pharma.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 空对象时停止
pharma_raw_codes=pharma['raw_code'].tolist() # 保留HDF原始代码用于高效行筛选
financial=pd.read_hdf(financial_path,key='financial_data',where='order_book_id in pharma_raw_codes',columns=['order_book_id','quarter','info_date','operating_revenue','profit_from_operation','net_accts_receivable']).rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','operating_revenue':'revenue','profit_from_operation':'operate_profit','net_accts_receivable':'accounts_receiv'}) # 按对象读取并标准化财报字段
financial['ts_code']=financial['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False) # 统一财报证券代码后缀
financial['end_date']=pd.PeriodIndex(financial['end_date'].str.upper(),freq='Q').end_time.normalize() # 将季度键转换为报告期末日期
if not basic_fields<=set(basic.columns) or not financial_fields<=set(financial.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 标准化后仍缺字段时停止
financial,version_audit=select_latest_disclosure(financial,['revenue','operate_profit','accounts_receiv']) # 统一保留最晚披露版本
panel=financial.merge(pharma[['ts_code','name','area']],on='ts_code',how='inner',validate='many_to_one').copy() # 合并公司标签if pharma.empty or panel.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 空对象或空连接时停止
if panel['end_date'].nunique()<4: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 期间不足时停止
panel=panel.loc[panel['revenue'].ne(0) & np.isfinite(panel[['revenue','operate_profit','accounts_receiv']]).all(axis=1)].copy() # 仅保留可定义比率的有限观测
if panel.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 有效分母观测为空时停止
panel['operating_margin']=panel['operate_profit']/panel['revenue'] # 计算营业利润率
panel['receivable_ratio']=panel['accounts_receiv']/panel['revenue'] # 计算应收账款收入比observed_periods=panel.groupby('ts_code')['end_date'].nunique() # 统计每家公司可用期间
eligible_codes=observed_periods.loc[observed_periods.ge(4)].index # 定义可观察同群
if len(eligible_codes)<2: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 同群不足时停止
eligible_panel=panel.loc[panel['ts_code'].isin(eligible_codes)].copy() # 固定候选同群
coverage=eligible_panel.groupby('end_date')['ts_code'].nunique().div(len(eligible_codes)) # 计算逐期覆盖率
coverage_threshold=0.80 # 课堂共同期覆盖情景，待业务批准
qualifying_dates=coverage.loc[coverage.ge(coverage_threshold)].index # 找到满足覆盖最低要求的候选期
if qualifying_dates.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查') # 无合格期时停止
latest_date=qualifying_dates.max() # 选择最新合格共同期
included_codes=set(eligible_panel.loc[eligible_panel['end_date'].eq(latest_date),'ts_code']) # 记录纳入公司
excluded_reasons={code:'合格期无有效财报' for code in set(eligible_codes)-included_codes} # 记录同群排除原因
excluded_reasons.update({code:'有效报告期少于4' for code in set(panel['ts_code'])-set(eligible_codes)}) # 记录观察不足原因
excluded_reasons.update({code:'无有效财报记录' for code in set(pharma['ts_code'])-set(panel['ts_code'])}) # 记录未进入观察面板的公司
cohort_audit={'candidate_date':latest_date,'coverage_threshold':coverage_threshold,'included':len(included_codes),'excluded':len(excluded_reasons),'excluded_reasons':excluded_reasons,'version_audit':version_audit} # 输出同群检查
latest=eligible_panel.loc[eligible_panel['end_date'].eq(latest_date)].nlargest(10,'revenue') # 生成合格同群最新期收入TOP10
trend=eligible_panel.groupby('end_date').agg(companies=('ts_code','nunique'),revenue=('revenue','sum'),operating_margin=('operating_margin','median'),receivable_ratio=('receivable_ratio','median')) # 输出同群趋势与分母
assert latest['revenue'].is_monotonic_decreasing # 核对排名方向
print(cohort_audit,latest[['ts_code','name','revenue','operating_margin','receivable_ratio']],trend) # 输出同群、排名与趋势case 路径是规定资产；用差额识别医保；单年月份直接制定备货[商业大数据分析与应用]